// Package compare px-table-checksum-go/compare/service.go
package compare

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"regexp"
	"sort" // 导入 sort 包用于排序
	"strings"
	"time"

	"px-table-checksum-go/config"
	"px-table-checksum-go/db" // 导入 db 包
	"px-table-checksum-go/models"
	"px-table-checksum-go/util" // 导入 util 包

	"github.com/sirupsen/logrus"
)

// 定义错误常量，便于统一处理和测试
var (
	// ErrNilDBManager 数据库管理器为空错误
	ErrNilDBManager = errors.New("数据库管理器为空")

	// ErrEmptySchemaOrTable 模式名或表名为空错误
	ErrEmptySchemaOrTable = errors.New("模式名或表名为空")

	// ErrIncompleteResults 校验结果不完整错误
	ErrIncompleteResults = errors.New("校验结果不完整")

	// ErrNilDB 数据库连接为空错误
	ErrNilDB = errors.New("数据库连接为空")

	// ErrNilDiffResult 差异结果为空错误
	ErrNilDiffResult = errors.New("差异结果为空")

	// ErrGetKeyInfo 获取键信息失败错误
	ErrGetKeyInfo = errors.New("获取表键信息失败")

	// ErrInvalidKeyInfo 无效键信息错误
	ErrInvalidKeyInfo = errors.New("无效的键信息")
)

// CompareService 表数据比较服务，负责对比源表和目标表数据差异
type CompareService struct {
	// 数据库连接管理器
	dbManager interface {
		GetSourceDB() *sql.DB
		GetTargetDB() *sql.DB
		GetChecksumDB() *sql.DB
	}

	// 日志字段，用于在日志中区分不同的比较任务
	logFields logrus.Fields

	// 配置选项
	config *config.AppConfig
}

// CompareOption 定义比较服务的配置选项函数类型
type CompareOption func(*CompareService)

// WithLogFields 设置日志字段选项
func WithLogFields(fields logrus.Fields) CompareOption {
	return func(s *CompareService) {
		s.logFields = fields
	}
}

// WithConfig 设置配置选项
func WithConfig(cfg *config.AppConfig) CompareOption {
	return func(s *CompareService) {
		s.config = cfg
	}
}

// NewCompareService 创建一个新的比较服务实例
// 使用选项模式允许灵活配置
func NewCompareService(opts ...CompareOption) (*CompareService, error) {
	dbManager := db.GetDBManager()
	if dbManager == nil {
		return nil, ErrNilDBManager
	}

	service := &CompareService{
		dbManager: dbManager,
		logFields: logrus.Fields{"component": "compare_service"},
		config:    &config.GlobalConfig,
	}

	// 应用所有配置选项
	for _, opt := range opts {
		opt(service)
	}

	return service, nil
}

// log 辅助方法：生成带有上下文字段的日志记录器
func (s *CompareService) log(ctx context.Context) *logrus.Entry {
	fields := logrus.Fields{}

	// 合并服务级日志字段
	for k, v := range s.logFields {
		fields[k] = v
	}

	// 获取上下文中可能存在的字段
	if ctx != nil {
		if reqID, ok := ctx.Value("request_id").(string); ok {
			fields["request_id"] = reqID
		}
	}

	return logrus.WithFields(fields)
}

// ValidateTableIdentifier 验证表标识符是否合法
func (s *CompareService) ValidateTableIdentifier(schemaName, tableName string) error {
	if schemaName == "" || tableName == "" {
		return fmt.Errorf("%w: schemaName=%q, tableName=%q",
			ErrEmptySchemaOrTable, schemaName, tableName)
	}
	return nil
}

// CompareTable 比较指定表的数据，支持上下文和取消操作
func (s *CompareService) CompareTable(ctx context.Context, schemaName, tableName string) (*models.CompareResult, error) {
	logger := s.log(ctx)
	logger.Infof("开始比较表 %s.%s 的数据...", schemaName, tableName)

	// 验证表标识符
	if err := s.ValidateTableIdentifier(schemaName, tableName); err != nil {
		return nil, err
	}

	// 获取表的校验结果
	sourceResults, targetResults, err := s.getTableChecksums(ctx, schemaName, tableName)
	if err != nil {
		return nil, fmt.Errorf("获取校验结果失败: %w", err)
	}

	// 检查是否存在校验结果
	// Modify: Improve incomplete result logging
	if len(sourceResults) == 0 {
		logger.Warnf("跳过表 %s.%s 比较，源库校验结果为空", schemaName, tableName)
		return nil, fmt.Errorf("%w: 源库校验结果为空 (%s.%s)", ErrIncompleteResults, schemaName, tableName)
	}
	if len(targetResults) == 0 {
		logger.Warnf("跳过表 %s.%s 比较，目标库校验结果为空", schemaName, tableName)
		return nil, fmt.Errorf("%w: 目标库校验结果为空 (%s.%s)", ErrIncompleteResults, schemaName, tableName)
	}
	logger.Debugf("获取校验结果完成: 源表 %d 条, 目标表 %d 条", len(sourceResults), len(targetResults))

	// 检查上下文是否已取消
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("比较操作被取消: %w", err)
	}

	// 检查CRC32是否有差异
	diffChunks, err := s.findDifferentChunks(ctx, sourceResults, targetResults)
	if err != nil {
		return nil, fmt.Errorf("查找差异块失败: %w", err)
	}

	// 如果没有差异，直接返回结果
	if len(diffChunks) == 0 {
		logger.Infof("表 %s.%s 所有数据块校验和匹配，表数据一致", schemaName, tableName)
		return &models.CompareResult{
			SchemaName:      schemaName,
			TableName:       tableName,
			TotalRowsSource: getTotalRows(sourceResults),
			TotalRowsTarget: getTotalRows(targetResults),
			MatchedRows:     getTotalRows(sourceResults), // 假设所有行都匹配
		}, nil
	}

	// 获取表的键信息
	// Note: Getting key info from source DB. Assumes target has the same key.
	keyInfo, err := db.GetBestUniqueKey(s.dbManager.GetSourceDB(), schemaName, tableName)
	if err != nil {
		logger.WithError(err).Errorf("获取表 %s.%s 的键信息失败", schemaName, tableName)
		return nil, fmt.Errorf("%w: %s.%s: %v",
			ErrGetKeyInfo, schemaName, tableName, err)
	}

	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		logger.Errorf("获取到表 %s.%s 的无效键信息", schemaName, tableName)
		return nil, fmt.Errorf("%w: 表 %s.%s 没有有效键信息", ErrInvalidKeyInfo, schemaName, tableName)
	}

	// 检查上下文是否已取消
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("比较操作被取消: %w", ctx.Err())
	}

	// 统计信息
	logger.Infof("表 %s.%s 发现 %d 个差异数据块，开始详细比较...", schemaName, tableName, len(diffChunks))

	// 合并所有差异块的比较结果
	finalResult := &models.CompareResult{
		SchemaName:      schemaName,
		TableName:       tableName,
		TotalRowsSource: getTotalRows(sourceResults),
		TotalRowsTarget: getTotalRows(targetResults),
		SourceOnlyRows:  make([]*models.TableRowData, 0, len(diffChunks)*10), // 预分配一定容量
		TargetOnlyRows:  make([]*models.TableRowData, 0, len(diffChunks)*10),
		DifferentRows:   make([]*models.RowDifference, 0, len(diffChunks)*10),
	}

	// 依次比较每个差异块
	for i, chunk := range diffChunks {
		// 周期性检查上下文是否已取消
		if i%5 == 0 { // 每5个块检查一次
			if err := ctx.Err(); err != nil {
				return nil, fmt.Errorf("比较操作被取消: %w", err)
			}
		}

		chunkLogger := logger.WithFields(logrus.Fields{
			"chunk_no": chunk.ChunkNo,
			"min_key":  chunk.MinKey,
			// Use source MaxKey as the chunk boundary for comparison if target MaxKey is empty
			"max_key_source": chunk.MaxKeySource,
			"max_key_target": chunk.MaxKeyTarget, // Log both max keys
		})
		chunkLogger.Info("开始详细比较块内容")

		chunkResult, err := s.compareChunk(ctx, schemaName, tableName, chunk, keyInfo)
		if err != nil {
			chunkLogger.WithError(err).Errorf("比较块 %d 失败", chunk.ChunkNo)
			// Record error but continue processing other chunks if possible
			// Depending on error type, could potentially stop processing this table.
			// For now, just log and continue with the next chunk.
			continue
		}

		if chunkResult == nil {
			chunkLogger.Warnf("比较块 %d 返回 nil 结果，跳过合并", chunk.ChunkNo)
			continue
		}

		// Merge results
		finalResult.OnlyInSource += chunkResult.OnlyInSource
		finalResult.OnlyInTarget += chunkResult.OnlyInTarget
		finalResult.MismatchedRows += chunkResult.MismatchedRows

		// Add detailed data, potentially limited by configuration
		maxDetailRows := 100 // Default limit
		if s.config != nil && s.config.Options.MaxDetailRows > 0 {
			maxDetailRows = s.config.Options.MaxDetailRows
		}

		// Append source-only rows up to the limit
		remainingSource := maxDetailRows - len(finalResult.SourceOnlyRows)
		if remainingSource > 0 && len(chunkResult.SourceOnlyRows) > 0 {
			rowsToAdd := chunkResult.SourceOnlyRows
			if len(rowsToAdd) > remainingSource {
				rowsToAdd = rowsToAdd[:remainingSource]
			}
			finalResult.SourceOnlyRows = append(finalResult.SourceOnlyRows, rowsToAdd...)
		}

		// Append target-only rows up to the limit
		remainingTarget := maxDetailRows - len(finalResult.TargetOnlyRows)
		if remainingTarget > 0 && len(chunkResult.TargetOnlyRows) > 0 {
			rowsToAdd := chunkResult.TargetOnlyRows
			if len(rowsToAdd) > remainingTarget {
				rowsToAdd = rowsToAdd[:remainingTarget]
			}
			finalResult.TargetOnlyRows = append(finalResult.TargetOnlyRows, rowsToAdd...)
		}

		// Append differing rows up to the limit
		remainingDiff := maxDetailRows - len(finalResult.DifferentRows)
		if remainingDiff > 0 && len(chunkResult.DifferentRows) > 0 {
			rowsToAdd := chunkResult.DifferentRows
			if len(rowsToAdd) > remainingDiff {
				rowsToAdd = rowsToAdd[:remainingDiff]
			}
			finalResult.DifferentRows = append(finalResult.DifferentRows, rowsToAdd...)
		}

		chunkLogger.Info("块详细比较完成")
	}

	// The CompareResult struct aggregates counts and stores detailed rows/diffs.
	// The final Result summary is printed after processing all chunks.

	return finalResult, nil
}

// getTableChecksums 获取表的校验结果
func (s *CompareService) getTableChecksums(ctx context.Context, schemaName, tableName string) ([]*models.ChecksumResult, []*models.ChecksumResult, error) {
	logger := s.log(ctx)

	// Validate table identifier
	if err := s.ValidateTableIdentifier(schemaName, tableName); err != nil {
		return nil, nil, err
	}

	// Get source checksum results
	logger.Debugf("获取源表校验结果: %s.%s (DBID: %s)", schemaName, tableName, s.config.Options.SourceDBID)
	sourceResults, err := db.GetChecksumResults(s.config.Options.SourceDBID, schemaName, tableName)
	if err != nil {
		// Modify: Improve error message to indicate source DB
		return nil, nil, fmt.Errorf("获取源表校验结果失败 (%s.%s): %w", schemaName, tableName, err)
	}
	logger.Debugf("获取到源表校验结果: %d 条", len(sourceResults))

	// Check context cancellation
	if err := ctx.Err(); err != nil {
		return nil, nil, fmt.Errorf("获取校验结果操作被取消: %w", err)
	}

	// Get target checksum results
	logger.Debugf("获取目标表校验结果: %s.%s (DBID: %s)", schemaName, tableName, s.config.Options.TargetDBID)
	targetResults, err := db.GetChecksumResults(s.config.Options.TargetDBID, schemaName, tableName)
	if err != nil {
		// Modify: Improve error message to indicate target DB
		return nil, nil, fmt.Errorf("获取目标表校验结果失败 (%s.%s): %w", schemaName, tableName, err)
	}
	logger.Debugf("获取到目标表校验结果: %d 条", len(targetResults))

	// No longer check for zero results here, as CompareTable handles it.

	return sourceResults, targetResults, nil
}

// findDifferentChunks 查找CRC32校验和不一致的数据块
// 对比源表和目标表的校验结果，找出不一致的数据块
func (s *CompareService) findDifferentChunks(ctx context.Context, sourceResults, targetResults []*models.ChecksumResult) ([]*models.DiffChunk, error) {
	logger := s.log(ctx)
	logger.WithFields(logrus.Fields{
		"source_chunks": len(sourceResults),
		"target_chunks": len(targetResults),
	}).Debug("开始比较源表和目标表的校验结果")

	// Create map indexes for quick lookup by ChunkNo
	sourceIndex := make(map[int]*models.ChecksumResult, len(sourceResults))
	for _, result := range sourceResults {
		sourceIndex[result.ChunkNo] = result
		// Log chunk info for debugging if debug level is enabled
		if logrus.GetLevel() >= logrus.DebugLevel {
			logger.WithFields(logrus.Fields{
				"chunk_no": result.ChunkNo, "min_key": result.MinKey, "max_key": result.MaxKey,
				"crc32": result.CRC32, "rows": result.RowCount, "db_id": result.DBID,
			}).Debug("Source Checksum Result")
		}
	}

	targetIndex := make(map[int]*models.ChecksumResult, len(targetResults))
	for _, result := range targetResults {
		targetIndex[result.ChunkNo] = result
		// Log chunk info for debugging if debug level is enabled
		if logrus.GetLevel() >= logrus.DebugLevel {
			logger.WithFields(logrus.Fields{
				"chunk_no": result.ChunkNo, "min_key": result.MinKey, "max_key": result.MaxKey,
				"crc32": result.CRC32, "rows": result.RowCount, "db_id": result.DBID,
			}).Debug("Target Checksum Result")
		}
	}

	// Identify differing chunks
	diffChunks := make([]*models.DiffChunk, 0, len(sourceIndex)/10+1) // Pre-allocate some capacity

	// Check context cancellation periodically
	checkInterval := 100 // check every 100 chunks

	// Iterate through all chunk numbers found in either source or target results
	allChunkNumbers := make(map[int]struct{})
	for chunkNo := range sourceIndex {
		allChunkNumbers[chunkNo] = struct{}{}
	}
	for chunkNo := range targetIndex {
		allChunkNumbers[chunkNo] = struct{}{}
	}

	chunkIndex := 0
	// Iterate through chunk numbers in ascending order for predictable logging
	sortedChunkNumbers := make([]int, 0, len(allChunkNumbers))
	for chunkNo := range allChunkNumbers {
		sortedChunkNumbers = append(sortedChunkNumbers, chunkNo)
	}
	sort.Ints(sortedChunkNumbers)

	for _, chunkNo := range sortedChunkNumbers {
		chunkIndex++
		if chunkIndex%checkInterval == 0 {
			if err := ctx.Err(); err != nil {
				logger.Warnf("查找差异块操作被取消在 chunk %d: %v", chunkNo, ctx.Err())
				return nil, fmt.Errorf("查找差异块操作被取消: %w", ctx.Err())
			}
		}

		sourceResult, sourceExists := sourceIndex[chunkNo]
		targetResult, targetExists := targetIndex[chunkNo]

		// Case 1: Chunk exists only in source
		if sourceExists && !targetExists {
			logger.WithFields(logrus.Fields{
				"chunk_no": chunkNo, "reason": "目标端缺少块",
				"source_crc32": sourceResult.CRC32, "source_rows": sourceResult.RowCount,
				"min_key": sourceResult.MinKey, "max_key": sourceResult.MaxKey,
			}).Warn("发现不一致的数据块")

			diffChunks = append(diffChunks, &models.DiffChunk{
				SchemaName: sourceResult.SchemaName, TableName: sourceResult.TableName, ChunkNo: chunkNo,
				MinKey: sourceResult.MinKey, MaxKeySource: sourceResult.MaxKey, MaxKeyTarget: "", // MaxKeyTarget is empty
				CRC32Source: sourceResult.CRC32, CRC32Target: 0, // CRC32Target is 0
				RowCountSource: sourceResult.RowCount, RowCountTarget: 0, // RowCountTarget is 0
			})
			continue // Move to next chunk number
		}

		// Case 2: Chunk exists only in target
		if !sourceExists && targetExists {
			logger.WithFields(logrus.Fields{
				"chunk_no": chunkNo, "reason": "源端缺少块",
				"target_crc32": targetResult.CRC32, "target_rows": targetResult.RowCount,
				"min_key": targetResult.MinKey, "max_key": targetResult.MaxKey,
			}).Warn("发现不一致的数据块")

			diffChunks = append(diffChunks, &models.DiffChunk{
				SchemaName: targetResult.SchemaName, TableName: targetResult.TableName, ChunkNo: chunkNo,
				MinKey: targetResult.MinKey, MaxKeySource: "", // MaxKeySource is empty
				MaxKeyTarget: targetResult.MaxKey, CRC32Source: 0, // CRC32Source is 0
				CRC32Target: targetResult.CRC32, RowCountSource: 0, // RowCountSource is 0
				RowCountTarget: targetResult.RowCount,
			})
			continue // Move to next chunk number
		}

		// Case 3: Chunk exists in both, check for differences
		if sourceResult.CRC32 != targetResult.CRC32 ||
			sourceResult.RowCount != targetResult.RowCount ||
			sourceResult.MaxKey != targetResult.MaxKey {

			logger.WithFields(logrus.Fields{
				"chunk_no": chunkNo, "reason": "属性不匹配",
				"source_crc32": sourceResult.CRC32, "target_crc32": targetResult.CRC32,
				"source_rows": sourceResult.RowCount, "target_rows": targetResult.RowCount,
				"source_max_key": sourceResult.MaxKey, "target_max_key": targetResult.MaxKey,
				"min_key": sourceResult.MinKey,
			}).Warn("发现不一致的数据块")

			diffChunks = append(diffChunks, &models.DiffChunk{
				SchemaName: sourceResult.SchemaName, TableName: sourceResult.TableName, ChunkNo: chunkNo,
				MinKey: sourceResult.MinKey, MaxKeySource: sourceResult.MaxKey, MaxKeyTarget: targetResult.MaxKey,
				CRC32Source: sourceResult.CRC32, CRC32Target: targetResult.CRC32,
				RowCountSource: sourceResult.RowCount, RowCountTarget: targetResult.RowCount,
			})
		} else {
			// Case 4: Chunk matches (CRC, RowCount, MaxKey are all the same)
			logger.WithFields(logrus.Fields{
				"chunk_no": chunkNo, "reason": "匹配",
				"crc32": sourceResult.CRC32, "rows": sourceResult.RowCount,
				"min_key": sourceResult.MinKey, "max_key": sourceResult.MaxKey,
			}).Debug("数据块完全匹配")
		}
	}

	logger.WithFields(logrus.Fields{
		"source_chunks_total": len(sourceResults),
		"target_chunks_total": len(targetResults),
		"diff_chunks_count":   len(diffChunks),
	}).Info("完成差异块查找")

	// Log detailed info for diff chunks if debug is enabled
	if len(diffChunks) > 0 && logrus.GetLevel() >= logrus.DebugLevel {
		logger.Debug("差异块详细信息:")
		for i, chunk := range diffChunks {
			logger.WithFields(logrus.Fields{
				"index": i, "chunk_no": chunk.ChunkNo, "min_key": chunk.MinKey,
				"max_key_source": chunk.MaxKeySource, "max_key_target": chunk.MaxKeyTarget,
				"crc32_source": chunk.CRC32Source, "crc32_target": chunk.CRC32Target,
				"rows_source": chunk.RowCountSource, "rows_target": chunk.RowCountTarget,
			}).Debug("差异块详情")
		}
	} else if len(diffChunks) == 0 {
		logger.Info("未发现任何差异块，两表数据一致")
	}

	return diffChunks, nil
}

// compareChunk 比较单个数据块的详细差异
// 获取源和目标的块数据，并进行行级别的详细比较
// 修改了 fetchChunkData 的调用方式，传递了 MinKey 和 MaxKey
func (s *CompareService) compareChunk(ctx context.Context, schemaName, tableName string, diffChunk *models.DiffChunk, keyInfo *db.KeyInfo) (*models.CompareResult, error) {
	logger := s.log(ctx).WithFields(logrus.Fields{
		"chunk_no": diffChunk.ChunkNo,
		"min_key":  diffChunk.MinKey,
		// Log both max keys from the diff chunk
		"max_key_source": diffChunk.MaxKeySource,
		"max_key_target": diffChunk.MaxKeyTarget,
	})
	logger.Info("开始比较数据块")

	// Parameter validation
	if diffChunk == nil {
		return nil, errors.New("差异块参数为空")
	}

	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		return nil, ErrInvalidKeyInfo // Use defined error constant
	}

	if err := s.ValidateTableIdentifier(schemaName, tableName); err != nil {
		return nil, err
	}

	// Check context cancellation
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("比较数据块操作被取消: %w", ctx.Err())
	}

	// Get source data
	logger.Debug("获取源数据块内容")
	sourceDB := s.dbManager.GetSourceDB() // Get DB connection pool
	if sourceDB == nil {
		return nil, fmt.Errorf("源%w", ErrNilDB)
	}

	// fetch data from source using source's max key as the upper boundary
	// Rename db parameter to dbConn to avoid shadowing the package name
	sourceRows, err := s.fetchChunkData(ctx, sourceDB, schemaName, tableName, diffChunk.MinKey, diffChunk.MaxKeySource, keyInfo)
	if err != nil {
		return nil, fmt.Errorf("获取源数据失败: %w", err)
	}
	logger.Debugf("源数据块获取完成, 行数: %d", len(sourceRows))

	// Check context cancellation
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("比较数据块操作被取消: %w", ctx.Err())
	}

	// Get target data
	logger.Debug("获取目标数据块内容")
	targetDB := s.dbManager.GetTargetDB() // Get DB connection pool
	if targetDB == nil {
		return nil, fmt.Errorf("目标%w", ErrNilDB)
	}

	// fetch data from target using target's max key as the upper boundary
	// Rename db parameter to dbConn to avoid shadowing the package name
	targetRows, err := s.fetchChunkData(ctx, targetDB, schemaName, tableName, diffChunk.MinKey, diffChunk.MaxKeyTarget, keyInfo)
	if err != nil {
		return nil, fmt.Errorf("获取目标数据失败: %w", err)
	}
	logger.Debugf("目标数据块获取完成, 行数: %d", len(targetRows))

	logger.WithFields(logrus.Fields{
		"source_rows": len(sourceRows),
		"target_rows": len(targetRows),
	}).Info("获取到数据行")

	// Check context cancellation
	if err := ctx.Err(); err != nil {
		return nil, fmt.Errorf("比较数据块操作被取消: %w", ctx.Err())
	}

	// Validate key columns
	if len(keyInfo.Columns) == 0 {
		return nil, errors.New("键列为空，无法进行比较")
	}

	// Convert to TableRowData
	// Note: Passing keyInfo.Columns ensures TableRowData knows which columns form the key.
	// NewTableRowData makes copies internally.
	sourceRowData := make([]*models.TableRowData, len(sourceRows))
	for i, row := range sourceRows {
		sourceRowData[i] = models.NewTableRowData(schemaName, tableName, row, keyInfo.Columns)
	}

	targetRowData := make([]*models.TableRowData, len(targetRows))
	for i, row := range targetRows {
		targetRowData[i] = models.NewTableRowData(schemaName, tableName, row, keyInfo.Columns)
	}

	// Compare row data
	logger.Debug("开始比较行数据")
	// models.CompareRows sorts internally and uses util.FormatForCRC for comparison
	result := models.CompareRows(schemaName, tableName, sourceRowData, targetRowData)

	// Output more detailed diff info
	logger.WithFields(logrus.Fields{
		"source_only_rows_count": result.OnlyInSource,
		"target_only_rows_count": result.OnlyInTarget,
		"different_rows_count":   result.MismatchedRows,
		"matched_rows_count":     result.MatchedRows, // Also log matched count
	}).Info("数据块比较结果统计")

	// Log inconsistencies between CRC and row comparison if any
	// Check if CRC matches but row comparison found differences
	if (result.OnlyInSource > 0 || result.OnlyInTarget > 0 || result.MismatchedRows > 0) &&
		diffChunk.CRC32Source == diffChunk.CRC32Target {
		logger.WithFields(logrus.Fields{
			"chunk_no":    diffChunk.ChunkNo,
			"crc32":       diffChunk.CRC32Source,
			"source_rows": len(sourceRowData),
			"target_rows": len(targetRowData),
		}).Warn("警告: 数据块CRC32匹配，但找到了具体差异行！这可能指示 CRC 计算逻辑或 FormatForCRC 函数存在问题。")
	} else if result.OnlyInSource == 0 && result.OnlyInTarget == 0 && result.MismatchedRows == 0 &&
		diffChunk.CRC32Source != diffChunk.CRC32Target {
		// Check if CRC differs but row comparison found no differences
		logger.WithFields(logrus.Fields{
			"chunk_no":     diffChunk.ChunkNo,
			"crc32_source": diffChunk.CRC32Source,
			"crc32_target": diffChunk.CRC32Target,
			"source_rows":  len(sourceRowData),
			"target_rows":  len(targetRowData),
		}).Warn("警告: 数据块CRC32不同，但未找到具体差异行！这强烈指示 CRC 计算逻辑或 FormatForCRC 函数存在问题，或某些难以比较的特殊值差异。")
		// Output sample row data for debugging if this happens
		if logrus.GetLevel() >= logrus.DebugLevel && (len(sourceRowData) > 0 || len(targetRowData) > 0) {
			if len(sourceRowData) > 0 {
				logger.WithField("sample_source_row_data", sourceRowData[0].Data).Debug("源数据块样例 (第1行)")
			}
			if len(targetRowData) > 0 {
				logger.WithField("sample_target_row_data", targetRowData[0].Data).Debug("目标数据块样例 (第1行)")
			}
		}
	}

	logger.WithField("summary", result.GetSummary()).Info("数据块比较完成")
	return result, nil
}

// fetchChunkData 获取指定块的表数据
// 执行SQL查询获取块内的表行数据
// Rename db parameter to dbConn to avoid shadowing the package name
func (s *CompareService) fetchChunkData(ctx context.Context, dbConn *sql.DB, schemaName, tableName, minKey, maxKey string, keyInfo *db.KeyInfo) ([]map[string]interface{}, error) {
	logger := s.log(ctx).WithFields(logrus.Fields{
		"schema":   schemaName,
		"table":    tableName,
		"min_key":  minKey,
		"max_key":  maxKey,
		"key_cols": keyInfo.Columns,
	})

	// Parameter validation
	if dbConn == nil {
		return nil, ErrNilDB
	}

	if err := s.ValidateTableIdentifier(schemaName, tableName); err != nil {
		return nil, err
	}

	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		return nil, ErrInvalidKeyInfo // Use defined error constant
	}

	// Build WHERE clause
	// Modify: Use util.KEY_PART_SEPARATOR for parsing min/max key strings
	whereClause, args, err := s.buildWhereClause(minKey, maxKey, keyInfo)
	if err != nil {
		return nil, fmt.Errorf("构建查询条件失败: %w", err)
	}

	// Build SQL query
	// Select all columns (*) as row comparison needs all data.
	query := fmt.Sprintf("SELECT * FROM `%s`.`%s`", schemaName, tableName)
	if whereClause != "" {
		query += " WHERE " + whereClause
	}

	// Order by key columns (Crucial for row comparison using sorted lists)
	if len(keyInfo.Columns) > 0 {
		orderByCols := make([]string, len(keyInfo.Columns))
		for i, col := range keyInfo.Columns {
			orderByCols[i] = fmt.Sprintf("`%s` ASC", col)
		}
		query += " ORDER BY " + strings.Join(orderByCols, ", ")
	}

	// Limit query rows to avoid excessive memory usage per chunk fetch
	limit := 2000 // Default limit
	// TODO: Use config.GlobalConfig.Options.ChunkSize. Should be >= ChunkSize used in source worker.
	query += fmt.Sprintf(" LIMIT %d", limit)

	// Log detailed SQL for debugging
	logger.WithFields(logrus.Fields{
		"query": query,
		"args":  fmt.Sprintf("%v", args),
	}).Debug("执行块数据查询")

	// Create context with timeout for the query
	queryCtx, cancel := context.WithTimeout(ctx, 30*time.Second) // 30 sec timeout, configurable?
	defer cancel()

	// Execute the query using dbConn
	rows, err := dbConn.QueryContext(queryCtx, query, args...)
	if err != nil {
		// Check for context cancellation error
		if queryCtx.Err() == context.DeadlineExceeded {
			logger.Errorf("查询块数据超时: %v", err)
			return nil, fmt.Errorf("查询块数据超时: %w", err)
		}
		logger.WithError(err).Errorf("查询块数据失败")
		return nil, fmt.Errorf("查询数据失败: %w", err)
	}
	defer rows.Close()

	// Get column names
	columns, err := rows.Columns()
	if err != nil {
		logger.WithError(err).Error("获取列名失败")
		return nil, fmt.Errorf("获取列名失败: %w", err)
	}
	if len(columns) == 0 {
		logger.Debug("查询返回0列，可能表不存在或没有列")      // Change to Debug
		return []map[string]interface{}{}, nil // Return empty slice if no columns/rows
	}

	// Get column types for scanning
	colTypes, err := rows.ColumnTypes()
	if err != nil {
		logger.WithError(err).Error("获取列类型失败")
		return nil, fmt.Errorf("获取列类型失败: %w", err)
	}

	// Prepare slice of pointers to scan targets
	scanArgs := make([]interface{}, len(columns))
	for i, colType := range colTypes {
		// Call db.CreateScanTarget using the db package name
		scanTarget := db.CreateScanTarget(colType.DatabaseTypeName())
		if scanTarget == nil {
			// This indicates an unhandled type in db.CreateScanTarget
			errMsg := fmt.Sprintf("无法为列 %s (类型 %s) 创建扫描目标", columns[i], colType.DatabaseTypeName())
			logger.Error(errMsg)
			// Return error for data integrity.
			return nil, errors.New(errMsg)
		}
		scanArgs[i] = scanTarget
	}

	// Prepare result slice
	// Pre-allocate capacity based on the limit
	result := make([]map[string]interface{}, 0, limit)

	// Iterate through the result set
	rowCount := 0
	for rows.Next() {
		// Check context cancellation periodically
		if rowCount%100 == 0 && ctx.Err() != nil {
			logger.Warn("获取数据操作被取消")
			return nil, fmt.Errorf("获取数据操作被取消: %w", ctx.Err())
		}

		// Scan the current row using scanArgs
		if err := rows.Scan(scanArgs...); err != nil {
			logger.WithError(err).Errorf("扫描行数据失败 (第 %d 行)", rowCount+1)
			return nil, fmt.Errorf("扫描行数据失败 (第 %d 行): %w", rowCount+1, err)
		}

		// Create row data map by getting values from the scanned targets
		rowData := make(map[string]interface{}, len(columns))
		for i, col := range columns {
			// Call db.GetScannedValue using the db package name
			val, scanErr := db.GetScannedValue(scanArgs[i]) // Correctly gets value from the pointer in scanArgs[i]
			if scanErr != nil {
				// Handle error getting scanned value
				errMsg := fmt.Sprintf("获取列 %s 的扫描值失败: %v", columns[i], scanErr)
				logger.Error(errMsg)
				return nil, errors.New(errMsg)
			}
			rowData[col] = val // Store the actual Go value
		}

		result = append(result, rowData)
		rowCount++
	}

	// Check for errors during iteration
	if err = rows.Err(); err != nil {
		logger.WithError(err).Error("遍历结果集错误")
		return nil, fmt.Errorf("遍历结果集错误: %w", err)
	}

	logger.WithField("row_count", len(result)).Debug("获取数据块完成")
	return result, nil
}

// buildWhereClause 构建SQL查询的WHERE条件
// 支持单键和复合键的条件构建
// Modify: Use util.KEY_PART_SEPARATOR for parsing min/max key strings
func (s *CompareService) buildWhereClause(minKey, maxKey string, keyInfo *db.KeyInfo) (string, []interface{}, error) {
	whereClause := ""
	args := make([]interface{}, 0)

	// Parse minKey (supports composite keys)
	// Condition: (col1, col2, ...) > (min_val1, min_val2, ...)
	if len(keyInfo.Columns) > 0 && minKey != "" {
		// Modify: Use util.ParseKeyFromString with util.KEY_PART_SEPARATOR
		minKeyValues, err := util.ParseKeyFromString(minKey, len(keyInfo.Columns))
		if err != nil {
			// Log warning but proceed with a simpler single-column comparison?
			// Or return error? Returning error is safer for data integrity.
			logrus.WithError(err).Warnf("解析最小键 '%s' 失败", minKey) // Log the error
			return "", nil, fmt.Errorf("解析最小键 '%s' 失败: %w", minKey, err)
		}

		// Use the slice from ParseKeyFromString directly as SQL arguments.
		minKeyArgs := minKeyValues

		// Check if parsed values match the expected number of key columns
		if len(minKeyArgs) != len(keyInfo.Columns) {
			// This indicates an issue with the stored minKey string format or schema mismatch.
			logrus.Errorf("键列数(%d)与解析后的minKey值数量(%d)不匹配: '%s'",
				len(keyInfo.Columns), len(minKeyArgs), minKey)
			return "", nil, fmt.Errorf("键列数(%d)与解析后的minKey值数量(%d)不匹配: '%s'",
				len(keyInfo.Columns), len(minKeyArgs), minKey)
		}

		// Build composite key "greater than" condition
		// (col1 > val1) OR (col1 = val1 AND col2 > val2) OR ...
		conditions := make([]string, 0, len(keyInfo.Columns))
		for i := 0; i < len(keyInfo.Columns); i++ {
			// Build the prefix equality conditions: (col1=v1 AND col2=v2 AND ... AND coli=vi)
			equalParts := make([]string, i)
			for j := 0; j < i; j++ {
				// Handle potential NULL values in the key parts
				if minKeyArgs[j] == nil {
					equalParts[j] = fmt.Sprintf("`%s` IS NULL", keyInfo.Columns[j])
					// No argument needed for IS NULL
				} else {
					equalParts[j] = fmt.Sprintf("`%s` = ?", keyInfo.Columns[j])
					args = append(args, minKeyArgs[j]) // Add non-NULL value as argument
				}
			}

			// Build the current column greater than condition: coli > vali
			greaterThanPart := ""
			// Handle potential NULL value for the current key part
			if minKeyArgs[i] == nil {
				// Cannot use '>' comparison with NULL. This scenario might need special handling
				// depending on how NULLs are ordered in the database and intended logic.
				// Percona Toolkit's pt-table-checksum treats NULL as the lowest possible value.
				// A condition like `(col1, col2) > (NULL, 'a')` is complex.
				// For simplicity here, let's assume keys are NOT NULL or this case implies the end of a NULL-prefixed range.
				// If NULL keys are possible and need precise handling, this logic needs refinement.
				// For now, log a warning and potentially skip this part of the condition.
				// A common approach for `> NULL` is to exclude the row if the key part is NULL.
				// Let's assume `> ?` with a NULL argument works as intended by the driver/DB (often evaluates to UNKNOWN/false).
				greaterThanPart = fmt.Sprintf("`%s` > ?", keyInfo.Columns[i])
				args = append(args, minKeyArgs[i]) // Add NULL as argument
				logrus.Warnf("构建 WHERE 条件时遇到 NULL 键值用于 '>' 比较: 列='%s', 键值=%v", keyInfo.Columns[i], minKeyArgs[i])

			} else {
				greaterThanPart = fmt.Sprintf("`%s` > ?", keyInfo.Columns[i])
				args = append(args, minKeyArgs[i]) // Add non-NULL value as argument
			}

			if i == 0 {
				// First column: simply col1 > val1
				// If the first part is NULL, the condition `col1 > NULL` is likely false.
				// A more robust solution for `(NULL, ...) > (NULL, ...)` would involve
				// `(col1 IS NULL AND col2 > val2) OR (col1 = NULL AND col2 = val2 AND col3 > val3) OR ...`
				// But sticking to the simplified structure.
				conditions = append(conditions, greaterThanPart)
			} else {
				// Subsequent columns: (prefix equality AND current column greater than)
				prefixCondition := strings.Join(equalParts, " AND ")
				if prefixCondition != "" { // Only include prefix if there are equality parts
					conditions = append(conditions, fmt.Sprintf("(%s AND %s)", prefixCondition, greaterThanPart))
				} else {
					// If no prefix equality parts (i=1, no col0 = val0), just the greaterThanPart
					conditions = append(conditions, greaterThanPart)
				}
			}
		}
		// Wrap the combined OR conditions in parentheses
		if len(conditions) > 0 {
			whereClause = "(" + strings.Join(conditions, " OR ") + ")"
		}

	}

	// Parse maxKey (supports composite keys)
	// Condition: (col1, col2, ...) <= (max_val1, max_val2, ...)
	if len(keyInfo.Columns) > 0 && maxKey != "" {
		// Modify: Use util.ParseKeyFromString with util.KEY_PART_SEPARATOR
		maxKeyValues, err := util.ParseKeyFromString(maxKey, len(keyInfo.Columns))
		if err != nil {
			logrus.WithError(err).Warnf("解析最大键 '%s' 失败", maxKey) // Log the error
			return "", nil, fmt.Errorf("解析最大键 '%s' 失败: %w", maxKey, err)
		}
		maxKeyArgs := maxKeyValues

		// Check if parsed values match the expected number of key columns
		if len(maxKeyArgs) != len(keyInfo.Columns) {
			logrus.Errorf("键列数(%d)与解析后的maxKey值数量(%d)不匹配: '%s'",
				len(keyInfo.Columns), len(maxKeyArgs), maxKey)
			return "", nil, fmt.Errorf("键列数(%d)与解析后的maxKey值数量(%d)不匹配: '%s'",
				len(keyInfo.Columns), len(maxKeyArgs), maxKey)
		}

		// Build composite key "less than or equal to" condition
		// col1 <= val1 AND col2 <= val2 AND ...
		maxConds := make([]string, len(keyInfo.Columns))
		for i, col := range keyInfo.Columns {
			// Handle potential NULL values in the key parts for <= comparison
			if maxKeyArgs[i] == nil {
				// col <= NULL is likely false. col IS NULL is true for NULL.
				// To include rows where key part is NULL and <= the max key value (which is NULL),
				// you'd need `col IS NULL`.
				// For simplicity, let's assume `col <= ?` with NULL argument works as intended (includes IS NULL).
				// A more robust solution for `(col1, col2) <= (NULL, ...)` would be complex.
				maxConds[i] = fmt.Sprintf("`%s` <= ?", col)
				args = append(args, maxKeyArgs[i]) // Add NULL as argument
				logrus.Warnf("构建 WHERE 条件时遇到 NULL 键值用于 '<=' 比较: 列='%s', 键值=%v", keyInfo.Columns[i], maxKeyArgs[i])

			} else {
				maxConds[i] = fmt.Sprintf("`%s` <= ?", col)
				args = append(args, maxKeyArgs[i]) // Add non-NULL value as argument
			}

		}
		maxWhereClause := "(" + strings.Join(maxConds, " AND ") + ")" // Wrap in parentheses

		// Combine with the minKey condition if it exists
		if whereClause != "" {
			whereClause += " AND " + maxWhereClause
		} else {
			whereClause = maxWhereClause
		}
	}

	// If both minKey and maxKey are empty, return empty clause and args
	// This scenario is unlikely for chunks > 1 based on how minKey/maxKey are generated.
	if minKey == "" && maxKey == "" {
		// This would mean fetching the entire table without partitioning?
		// The BuildChunkSQL query template already includes `> (?)` for the first chunk.
		// So minKey should effectively always be non-empty after the first chunk's MaxKey is determined.
		// If minKey is empty AND maxKey is empty, it should imply the first chunk.
		// But the SQL template has `> (?)`
		// Re-evaluating the initial SQL template and chunking logic:
		// BuildChunkSQL creates SELECT ... WHERE (key_col) > (?) ORDER BY ... LIMIT 1000
		// prepareQueryParams for chunk 1 uses GetInitialStartKey("0"), which parses to ["0"]
		// fetchChunkData is called with minKey="0", maxKey=<first chunk's MaxKey>
		// buildWhereClause needs to handle minKey="0" and maxKey.
		// The `> "0"` part is handled by the minKey logic.
		// So, minKey will always be non-empty for all chunks except maybe a special initial state?
		// Let's remove the check for both being empty, as it should not happen with the current design.
		// If minKey is empty and maxKey is not, it's likely an error state or edge case in chunking.
		// If maxKey is empty and minKey is not, it's the last chunk (or an empty table case).

		// If the function is called with both minKey and maxKey being empty strings,
		// it would result in an empty whereClause here.
		// The original BuildChunkSQL template *always* has `WHERE (key_col) > (?)`,
		// implying minKey is *always* used as the lower bound (exclusive).
		// An empty minKey (formatted as "0" or "0,0") signifies the start.
		// An empty maxKey indicates the end of the range.
		// The only time there's no WHERE clause is likely an empty table scan,
		// but even then, the template has `> (?)`.
		// This case might not be reachable with the current chunking logic.
		// Let's assume it's not a valid state or needs specific handling elsewhere.
		// Returning empty clause/args is technically correct for no filter, but might contradict the SQL template structure.
		// Sticking to the logic derived from minKey/maxKey parsing. If both are empty, the above logic won't build conditions.
		// Returning empty is fine in that specific case, but it implies no filtering.

	}

	return whereClause, args, nil
}

// Auxiliary function: Checks if a byte slice might be a date/datetime format
// This helper is not strictly needed by FormatForCRC anymore as GetScannedValue
// is expected to return time.Time. Keeping it as it was part of the original code
// and might be useful elsewhere or as a fallback if GetScannedValue doesn't work as expected.
func isDateTime(b []byte) bool {
	// Implementation remains the same
	if b == nil || len(b) < 8 || len(b) > 30 {
		return false
	}
	s := string(b)
	datePatterns := []string{
		`^\d{4}-\d{2}-\d{2}$`,
		`^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}$`,
		`^\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{1,6}$`,
		`^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d{1,6})?Z?$`,
	}
	for _, pattern := range datePatterns {
		match, err := regexp.MatchString(pattern, s)
		if err != nil {
			continue
		}
		if match {
			return true
		}
	}
	return false
}

// Auxiliary function: Parses a string into a time object
// Similar to isDateTime, might be redundant if GetScannedValue works correctly.
// Keeping for now.
func parseDateTime(s string) (time.Time, error) {
	// Implementation remains the same
	if s == "" {
		return time.Time{}, errors.New("空日期时间字符串")
	}
	formats := []string{
		"2006-01-02 15:04:05.999999", "2006-01-02 15:04:05.999", "2006-01-02 15:04:05", "2006-01-02",
		time.RFC3339Nano, time.RFC3339,
	}
	var lastErr error
	for _, format := range formats {
		if t, err := time.Parse(format, s); err == nil {
			return t, nil
		} else {
			lastErr = err
		}
	}
	return time.Time{}, fmt.Errorf("无法解析日期时间 '%s': %w", s, lastErr)
}

// getTotalRows 计算结果集的总行数
func getTotalRows(results []*models.ChecksumResult) int {
	if results == nil {
		return 0
	}

	total := 0
	for _, result := range results {
		if result != nil {
			total += result.RowCount
		}
	}
	return total
}

// 以下是导出用于兼容旧代码的函数

// NewCompareChecksums 比较两个表的校验结果(兼容旧代码)
func NewCompareChecksums(schemaName, tableName string) ([]*models.DiffChunk, error) {
	logger := logrus.WithFields(logrus.Fields{
		"schemaName": schemaName,
		"tableName":  tableName,
		"func":       "NewCompareChecksums",
	})
	logger.Debug("开始比较校验和")

	// Create compare service instance
	service, err := NewCompareService()
	if err != nil {
		logger.WithError(err).Error("创建比较服务实例失败")
		return nil, fmt.Errorf("创建比较服务实例失败: %w", err)
	}
	logger.Debug("成功创建比较服务实例")

	// Get source and target checksum results
	logger.Debug("开始获取源表和目标表的校验和结果")
	ctx := context.Background()
	sourceResults, targetResults, err := service.getTableChecksums(ctx, schemaName, tableName)
	if err != nil {
		// getTableChecksums already logs specific error for source/target
		return nil, fmt.Errorf("获取表校验结果失败: %w", err)
	}

	// Check for zero results
	if len(sourceResults) == 0 || len(targetResults) == 0 {
		// If getTableChecksums returned results = 0 but err = nil, it means no rows in checksum table.
		// CheckTableExistsInChecksum can confirm this.
		checksumDB := db.GetDBManager().GetChecksumDB()
		srcExists, errSrc := CheckTableExistsInChecksum(checksumDB, schemaName, tableName, config.GlobalConfig.Options.SourceDBID)
		tgtExists, errTgt := CheckTableExistsInChecksum(checksumDB, schemaName, tableName, config.GlobalConfig.Options.TargetDBID)

		// Log potential issues
		if errSrc != nil {
			logger.WithError(errSrc).Warnf("检查源校验数据存在性失败")
		}
		if errTgt != nil {
			logger.WithError(errTgt).Warnf("检查目标校验数据存在性失败")
		}

		// Provide more specific error message based on existence check
		if !srcExists && !tgtExists {
			logger.Warnf("表 %s.%s 在校验库中源端和目标端均无数据", schemaName, tableName)
			return nil, fmt.Errorf("表 %s.%s 在校验库中源端和目标端均无数据", schemaName, tableName)
		} else if !srcExists {
			logger.Warnf("表 %s.%s 在校验库中源端无数据", schemaName, tableName)
			return nil, fmt.Errorf("表 %s.%s 在校验库中源端无数据", schemaName, tableName)
		} else if !tgtExists {
			logger.Warnf("表 %s.%s 在校验库中目标端无数据", schemaName, tableName)
			return nil, fmt.Errorf("表 %s.%s 在校验库中目标端无数据", schemaName, tableName)
		}
		// If we reach here, it means Exists returned true but Get returned empty list.
		// This is strange, might indicate transient issue or data corruption.
		// Fallback to the original incomplete results error.
		logger.Warn("Checksum results were empty but existence check was positive. Incomplete results.")
		return nil, ErrIncompleteResults
	}

	logger.Infof("成功获取校验结果: 源表=%d条, 目标表=%d条", len(sourceResults), len(targetResults))

	// Find differing chunks
	logger.Debug("开始查找不同的数据块")
	diffChunks, err := service.findDifferentChunks(ctx, sourceResults, targetResults)
	if err != nil {
		// findDifferentChunks already logs the error
		return nil, fmt.Errorf("查找不同数据块失败: %w", err)
	}

	logger.Infof("比较完成，找到 %d 个不同的数据块", len(diffChunks))

	// Log first few diff chunks for debugging
	if len(diffChunks) > 0 && logrus.GetLevel() >= logrus.DebugLevel {
		count := min(3, len(diffChunks))
		logger.Debugf("前 %d 个不同数据块:", count)
		for i := 0; i < count; i++ {
			chunk := diffChunks[i]
			logger.Debugf("  - 块 #%d: ChunkNo=%d, MinKey=%s, 源CRC32=%d, 目标CRC32=%d, 源行数=%d, 目标行数=%d",
				i+1, chunk.ChunkNo, chunk.MinKey, chunk.CRC32Source, chunk.CRC32Target, chunk.RowCountSource, chunk.RowCountTarget)
		}
	}

	return diffChunks, nil
}

// NewFindDifferencesInChunkWithContext 带上下文的查找差异，支持取消和超时
func NewFindDifferencesInChunkWithContext(ctx context.Context, schemaName, tableName string, diffChunk *models.DiffChunk, keyInfo *db.KeyInfo) (*models.DiffResult, error) {
	// Check context cancellation
	select {
	case <-ctx.Done():
		return nil, ctx.Err()
	default:
		// Continue processing
	}

	// Create compare service instance
	service, err := NewCompareService(WithLogFields(logrus.Fields{"component": "find_differences"}))
	if err != nil {
		logrus.WithError(err).Error("创建比较服务失败")
		return nil, fmt.Errorf("创建比较服务实例失败: %w", err)
	}

	logger := logrus.WithFields(logrus.Fields{
		"schema":   schemaName,
		"table":    tableName,
		"chunk_no": diffChunk.ChunkNo,
		"function": "NewFindDifferencesInChunk",
	})
	logger.Debug("开始查找块内差异")

	// Log diff chunk info
	logger.WithFields(logrus.Fields{
		"min_key": diffChunk.MinKey, "max_key_source": diffChunk.MaxKeySource, "max_key_target": diffChunk.MaxKeyTarget,
		"crc32_source": diffChunk.CRC32Source, "crc32_target": diffChunk.CRC32Target,
		"rows_source": diffChunk.RowCountSource, "rows_target": diffChunk.RowCountTarget,
	}).Debug("处理差异块详情")

	// Log key info
	logger.WithFields(logrus.Fields{
		"key_name": keyInfo.Name, "key_columns": keyInfo.Columns, "is_primary": keyInfo.IsPrimary,
	}).Debug("使用键信息")

	// Execute compareChunk in a goroutine with timeout monitoring
	operationDone := make(chan struct{})
	var result *models.CompareResult
	var compareErr error

	go func() {
		defer close(operationDone)
		// Pass the context received by this function
		result, compareErr = service.compareChunk(ctx, schemaName, tableName, diffChunk, keyInfo)
	}()

	// Wait for the operation to complete or context cancellation/timeout
	select {
	case <-ctx.Done():
		// Operation cancelled or timed out, wait for the goroutine to finish
		logger.Warnf("比较块操作被取消或超时: %v", ctx.Err())
		<-operationDone // Wait for the goroutine to exit
		// check if compareErr was set before context cancellation
		if compareErr != nil && !errors.Is(compareErr, ctx.Err()) {
			// An error happened in compareChunk *before* context cancellation was detected
			return nil, fmt.Errorf("比较块 %d 数据失败 (操作已取消，原始错误: %v): %w", diffChunk.ChunkNo, compareErr, ctx.Err())
		}
		return nil, ctx.Err() // Return context error if no specific compare error occurred
	case <-operationDone:
		// Operation completed, check result
		if compareErr != nil {
			logger.WithError(compareErr).Errorf("执行块比较失败 chunk %d", diffChunk.ChunkNo)
			return nil, fmt.Errorf("比较块 %d 数据失败: %w", diffChunk.ChunkNo, compareErr)
		}
	}

	if result == nil {
		logger.Errorf("compareChunk返回了空结果 chunk %d", diffChunk.ChunkNo)
		return nil, fmt.Errorf("比较块 %d 结果为空", diffChunk.ChunkNo)
	}

	// Log comparison result details
	logger.WithFields(logrus.Fields{
		"source_only_rows_count": result.OnlyInSource, "target_only_rows_count": result.OnlyInTarget,
		"different_rows_count": result.MismatchedRows, "matched_rows_count": result.MatchedRows,
	}).Debug("比较结果详情")

	// Log sample data for debugging if needed
	if logrus.GetLevel() >= logrus.DebugLevel {
		if len(result.SourceOnlyRows) > 0 {
			// Check if SourceOnlyRows[0] is not nil before calling GetKey()
			if result.SourceOnlyRows[0] != nil {
				logger.WithFields(logrus.Fields{"row_key": result.SourceOnlyRows[0].GetKey()}).Debug("源表独有行样例")
			} else {
				logger.Debug("源表独有行列表不为空但第一个元素为nil")
			}
		}
		if len(result.TargetOnlyRows) > 0 {
			// Check if TargetOnlyRows[0] is not nil before calling GetKey()
			if result.TargetOnlyRows[0] != nil {
				logger.WithFields(logrus.Fields{"row_key": result.TargetOnlyRows[0].GetKey()}).Debug("目标表独有行样例")
			} else {
				logger.Debug("目标表独有行列表不为空但第一个元素为nil")
			}
		}
		if len(result.DifferentRows) > 0 {
			// Check if DifferentRows[0] is not nil and Key is not nil before logging
			if result.DifferentRows[0] != nil && result.DifferentRows[0].Key != "" {
				logger.WithFields(logrus.Fields{"row_key": result.DifferentRows[0].Key, "diff_columns_count": len(result.DifferentRows[0].DifferentColumns)}).Debug("不匹配行样例")
			} else {
				logger.Debug("不匹配行列表不为空但第一个元素为nil或键为nil")
			}
		}
	}

	// Convert CompareResult to DiffResult (for compatibility with old code)
	// Use helper functions defined below to convert data structures
	diffResult := &models.DiffResult{
		OnlyInSource:     result.OnlyInSource,
		OnlyInTarget:     result.OnlyInTarget,
		DifferentRows:    result.MismatchedRows,                      // Total count of mismatched rows
		SourceOnlyRows:   convertToMapSlice(result.SourceOnlyRows),   // Detailed source-only rows
		TargetOnlyRows:   convertToMapSlice(result.TargetOnlyRows),   // Detailed target-only rows
		DifferentDetails: convertToDiffDetails(result.DifferentRows), // Detailed mismatched rows
	}

	logger.WithFields(logrus.Fields{
		"source_only_count_agg": diffResult.OnlyInSource, "target_only_count_agg": diffResult.OnlyInTarget,
		"different_rows_count_agg": diffResult.DifferentRows, "different_details_len": len(diffResult.DifferentDetails),
		"source_only_rows_len": len(diffResult.SourceOnlyRows), "target_only_rows_len": len(diffResult.TargetOnlyRows),
	}).Debug("转换后的差异结果详情 (聚合计数 vs 详细列表长度)")

	// Consistency check after conversion and log warnings if counts mismatch detailed lists.
	// Auto-correct counts to match list lengths if they differ.
	// This might indicate issues in convertToMapSlice or convertToDiffDetails.
	if diffResult.OnlyInSource > 0 && len(diffResult.SourceOnlyRows) != diffResult.OnlyInSource {
		logger.Warnf("源端独有行计数不匹配 (统计 %d vs 详情 %d)，根据详情长度调整统计计数", diffResult.OnlyInSource, len(diffResult.SourceOnlyRows))
		diffResult.OnlyInSource = len(diffResult.SourceOnlyRows)
	}
	if diffResult.OnlyInTarget > 0 && len(diffResult.TargetOnlyRows) != diffResult.OnlyInTarget {
		logger.Warnf("目标端独有行计数不匹配 (统计 %d vs 详情 %d)，根据详情长度调整统计计数", diffResult.OnlyInTarget, len(diffResult.TargetOnlyRows))
		diffResult.OnlyInTarget = len(diffResult.TargetOnlyRows)
	}
	if diffResult.DifferentRows > 0 && len(diffResult.DifferentDetails) != diffResult.DifferentRows {
		logger.Warnf("内容不同行计数不匹配 (统计 %d vs 详情 %d)，根据详情长度调整统计计数", diffResult.DifferentRows, len(diffResult.DifferentDetails))
		diffResult.DifferentRows = len(diffResult.DifferentDetails)
	}

	// Final check on actual differences based on list lengths after potential adjustment
	actualDiffs := len(diffResult.SourceOnlyRows) + len(diffResult.TargetOnlyRows) + len(diffResult.DifferentDetails)
	if actualDiffs == 0 {
		logger.Info("转换后未发现实际差异数据 (详细列表为空)")
	} else {
		logger.Infof("转换后共发现 %d 条实际差异数据 (根据详细列表统计)", actualDiffs)
	}

	logger.Infof("完成查找块内差异, 源端独有=%d, 目标端独有=%d, 内容不同=%d (最终计数)",
		diffResult.OnlyInSource, diffResult.OnlyInTarget, diffResult.DifferentRows)

	return diffResult, nil
}

// CheckTableExistsInChecksum 检查Checksum表中是否存在指定表的结果
func CheckTableExistsInChecksum(dbConn *sql.DB, schemaName, tableName, dbID string) (bool, error) {
	// Create logger with fields for tracking
	logger := logrus.WithFields(logrus.Fields{
		"schemaName": schemaName,
		"tableName":  tableName,
		"dbID":       dbID,
		"func":       "CheckTableExistsInChecksum",
	})

	logger.Debug("开始检查表数据是否存在于校验库")

	// Check database connection
	if dbConn == nil {
		logger.Error("数据库连接为空")
		return false, ErrNilDB // Use defined error constant
	}

	// Parameter validation
	if schemaName == "" || tableName == "" || dbID == "" {
		logger.Error("无效的参数: schemaName、tableName或dbID为空")
		return false, errors.New("参数无效")
	}

	// Build SQL query
	query := `
		SELECT COUNT(*)
		FROM t_checksum
		WHERE f_dbid = ? AND f_schema_name = ? AND f_table_name = ?
	`

	// Log SQL and parameters
	logger.Debugf("执行SQL: %s", strings.ReplaceAll(query, "\n", " "))
	logger.Debugf("SQL参数: [%s, %s, %s]", dbID, schemaName, tableName)

	// Add timeout context
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) // 10 sec timeout
	defer cancel()

	// Execute query
	var count int
	err := dbConn.QueryRowContext(ctx, query, dbID, schemaName, tableName).Scan(&count)
	if err != nil {
		// Check if it was a context timeout error
		if ctx.Err() == context.DeadlineExceeded {
			logger.WithError(ctx.Err()).Errorf("检查表校验数据存在性超时")
			return false, fmt.Errorf("检查表校验数据存在性超时: %w", ctx.Err())
		}
		logger.WithError(err).Errorf("查询表校验数据失败")
		return false, fmt.Errorf("查询失败: %w", err)
	}

	// Log query result
	exists := count > 0
	if exists {
		logger.Debugf("找到匹配记录: %d 条", count)
	} else {
		logger.Debug("未找到任何匹配记录") // Change to Debug as finding no data is not always a warning
	}

	return exists, nil
}

// GenerateFixSQLs 根据差异生成修复SQL
// 修复点：在生成 INSERT SQL 时，需要处理各种数据类型到字符串的转换和转义，
//
//	特别是 BLOB/BINARY 类型需要使用 UNHEX 函数。
//	在生成 UPDATE/DELETE SQL 时， WHERE 子句的键值需要正确格式化和转义。
//	注意：generateInsertSQL, generateDeleteSQL, generateUpdateSQL 以及 escapeSQLValue 需要修改。
func GenerateFixSQLs(schemaName, tableName string, keyInfo *db.KeyInfo, allColsInfo []db.ColumnInfo, diffResult *models.DiffResult) ([]string, error) {
	if diffResult == nil {
		return nil, ErrNilDiffResult // Use defined error constant
	}

	logger := logrus.WithFields(logrus.Fields{
		"schema":                schemaName,
		"table":                 tableName,
		"source_only_agg":       diffResult.OnlyInSource, // Log aggregated counts from DiffResult
		"target_only_agg":       diffResult.OnlyInTarget,
		"different_rows_agg":    diffResult.DifferentRows,
		"source_only_len":       len(diffResult.SourceOnlyRows), // Log detailed list lengths
		"target_only_len":       len(diffResult.TargetOnlyRows),
		"different_details_len": len(diffResult.DifferentDetails),
		"function":              "GenerateFixSQLs",
	})

	logger.Info("开始生成修复SQL")
	logger.Debugf("输入参数: schemaName=%s, tableName=%s, keyInfo.Columns=%v, allColsInfo.len=%d",
		schemaName, tableName, keyInfo.Columns, len(allColsInfo))

	// Check key info validity for DELETE/UPDATE
	keyInfoValidForFix := keyInfo != nil && len(keyInfo.Columns) > 0

	var sqls []string // Slice to collect generated SQL statements

	// 1. Records only in source -> Generate INSERT
	// Iterate over the detailed list of source-only rows
	if len(diffResult.SourceOnlyRows) > 0 { // diffResult.SourceOnlyRows is []map[string]interface{}
		logger.Debugf("开始为 %d 条源表独有记录生成 INSERT SQL", len(diffResult.SourceOnlyRows))
		insertCount := 0
		for i, rowDataMap := range diffResult.SourceOnlyRows {
			if i < 3 || logrus.GetLevel() >= logrus.DebugLevel { // Log first 3 or all in debug
				logger.Debugf("源表独有行示例 #%d: %+v", i+1, rowDataMap)
			}
			// Call generateInsertSQL with the data map
			insertSQL := generateInsertSQL(schemaName, tableName, rowDataMap)
			if insertSQL != "" {
				sqls = append(sqls, insertSQL)
				insertCount++
			} else {
				// Log a warning if SQL generation failed for a specific row
				logger.Warnf("为源表独有行生成 INSERT SQL 失败，行数据: %+v", rowDataMap)
			}
		}
		logger.Infof("为 %d 条源表独有记录生成了 %d 条 INSERT SQL", len(diffResult.SourceOnlyRows), insertCount)
	} else {
		logger.Debug("没有源表独有记录，跳过生成 INSERT")
	}

	// 2. Records only in target -> Generate DELETE
	// Iterate over the detailed list of target-only rows
	if len(diffResult.TargetOnlyRows) > 0 { // diffResult.TargetOnlyRows is []map[string]interface{}
		logger.Debugf("开始为 %d 条目标表独有记录生成 DELETE SQL", len(diffResult.TargetOnlyRows))
		deleteCount := 0
		if !keyInfoValidForFix {
			// If key info is invalid, log error and skip DELETE generation
			logger.Error("未提供有效的键信息，无法为目标独有记录生成 DELETE SQL")
		} else {
			for i, rowDataMap := range diffResult.TargetOnlyRows {
				if i < 3 || logrus.GetLevel() >= logrus.DebugLevel { // Log first 3 or all in debug
					logger.Debugf("目标表独有行示例 #%d: %+v", i+1, rowDataMap)
				}
				// Call generateDeleteSQL with key info and data map
				deleteSQL := generateDeleteSQL(schemaName, tableName, keyInfo, rowDataMap)
				if deleteSQL != "" {
					sqls = append(sqls, deleteSQL)
					deleteCount++
				} else {
					logger.Warnf("为目标表独有行生成 DELETE SQL 失败，行数据: %+v", rowDataMap)
				}
			}
			logger.Infof("为 %d 条目标表独有记录生成了 %d 条 DELETE SQL", len(diffResult.TargetOnlyRows), deleteCount)
		}
	} else {
		logger.Debug("没有目标表独有记录，跳过生成 DELETE")
	}

	// 3. Records with same key but different content -> Generate UPDATE
	// Iterate over the detailed list of differing rows
	if len(diffResult.DifferentDetails) > 0 { // diffResult.DifferentDetails is []models.RowDiff
		logger.Debugf("开始为 %d 条内容不同的记录生成 UPDATE SQL", len(diffResult.DifferentDetails))
		updateCount := 0
		if !keyInfoValidForFix {
			// If key info is invalid, log error and skip UPDATE generation
			logger.Error("未提供有效的键信息，无法为内容不同记录生成 UPDATE SQL")
		} else {
			for i, diff := range diffResult.DifferentDetails {
				// Log sample diff details for debugging
				if i < 3 || logrus.GetLevel() >= logrus.DebugLevel {
					logger.Debugf("不同行示例 #%d: 键=%v, 不同列数=%d", i+1, diff.Key, len(diff.DiffColumns))
					count := 0
					// Log sample differing columns for this row
					for colName, colDiff := range diff.DiffColumns {
						if count < 5 || logrus.GetLevel() >= logrus.DebugLevel { // Log up to 5 diff columns or all in debug
							// Use fmt.Sprintf to show the values. For complex types like []byte, %v shows type and value.
							logger.Debugf("  - 列差异 #%d: 列名=%s, 源值=%v, 目标值=%v",
								count+1, colName, colDiff.SourceValue, colDiff.TargetValue)
							count++
						} else {
							break // Limit sample output
						}
					}
				}

				// Call generateUpdateSQL with key info and RowDiff
				updateSQL := generateUpdateSQL(schemaName, tableName, keyInfo, diff)
				if updateSQL != "" {
					sqls = append(sqls, updateSQL)
					updateCount++
				} else {
					logger.Warnf("为内容不同的行生成 UPDATE SQL 失败，键值: %v", diff.Key)
				}
			}
			logger.Infof("为 %d 条内容不同的记录生成了 %d 条 UPDATE SQL", len(diffResult.DifferentDetails), updateCount)
		}
	} else {
		logger.Debug("没有内容不同的记录，跳过生成 UPDATE")
	}

	// Check generated SQL count
	totalSQLs := len(sqls)
	if totalSQLs == 0 {
		// Warn if there were differences (according to aggregated counts) but no SQL generated.
		// This could happen if keyInfo was invalid for DELETE/UPDATE or conversion failed.
		if diffResult.OnlyInSource > 0 || diffResult.OnlyInTarget > 0 || diffResult.DifferentRows > 0 {
			logger.Warnf("有差异但未生成任何SQL! 源端独有=%d(详情:%d), 目标端独有=%d(详情:%d), 内容不同=%d(详情:%d)",
				diffResult.OnlyInSource, len(diffResult.SourceOnlyRows),
				diffResult.OnlyInTarget, len(diffResult.TargetOnlyRows),
				diffResult.DifferentRows, len(diffResult.DifferentDetails))

			if !keyInfoValidForFix {
				logger.Error("无法生成 DELETE/UPDATE SQL，因为键信息无效")
			}
			// Check if detailed lists are unexpectedly empty when aggregated counts > 0
			if diffResult.OnlyInSource > 0 && len(diffResult.SourceOnlyRows) == 0 {
				logger.Error("源端独有行计数 > 0 但详情列表为空")
			}
			if diffResult.OnlyInTarget > 0 && len(diffResult.TargetOnlyRows) == 0 {
				logger.Error("目标端独有行计数 > 0 但详情列表为空")
			}
			if diffResult.DifferentRows > 0 && len(diffResult.DifferentDetails) == 0 {
				logger.Error("内容不同行计数 > 0 但详情列表为空")
			}

		} else {
			// No differences found at all (both aggregated counts and detailed lists are empty)
			// This is the expected case for a consistent table.
			logger.Info("未生成任何SQL语句 (未发现差异)") // Change to Info level
		}
		return nil, nil
	}

	logger.Infof("成功生成 %d 条修复SQL", totalSQLs)
	return sqls, nil
}

// generateInsertSQL 生成插入SQL (修改以处理不同数据类型)
func generateInsertSQL(schemaName, tableName string, row map[string]interface{}) string {
	columns := make([]string, 0, len(row))
	values := make([]string, 0, len(row))

	// Iterate over row data map. Order is not guaranteed by map iteration.
	// To ensure stable SQL output for testing/diff, you might sort column names first.
	// For now, keeping it simple and iterating directly.

	for col, val := range row {
		columns = append(columns, "`"+col+"`")
		// Call escapeSQLValue function to format value
		values = append(values, escapeSQLValue(val))
	}

	if len(columns) == 0 {
		// Should not happen for non-empty rows passed from comparison logic, but handle defensively.
		logrus.Warnf("生成 INSERT SQL 失败: 表 %s.%s 行数据为空 %+v", schemaName, tableName, row)
		return ""
	}

	return fmt.Sprintf("INSERT INTO `%s`.`%s` (%s) VALUES (%s);",
		schemaName, tableName, strings.Join(columns, ", "), strings.Join(values, ", "))
}

// generateDeleteSQL 生成删除SQL (修改以处理键值格式化)
func generateDeleteSQL(schemaName, tableName string, keyInfo *db.KeyInfo, row map[string]interface{}) string {
	var whereParts []string

	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		logrus.Error("generateDeleteSQL: 键信息无效")
		return ""
	}

	// Build WHERE clause using key columns from keyInfo and values from the row map
	for _, col := range keyInfo.Columns {
		val, exists := row[col]
		if !exists {
			// Warning: Value for a key column is missing in the row data!
			logrus.Warnf("生成 DELETE SQL 失败: 表 %s.%s 行数据中缺少键列 '%s' 的值。行数据: %+v",
				schemaName, tableName, col, row)
			return "" // Cannot generate valid SQL without all key values
		}

		// Format key value using escapeSQLValue for SQL literal
		if val == nil {
			whereParts = append(whereParts, fmt.Sprintf("`%s` IS NULL", col))
			// No argument needed for IS NULL
		} else {
			// Use escapeSQLValue to format the value correctly for SQL literal
			whereParts = append(whereParts, fmt.Sprintf("`%s` = %s", col, escapeSQLValue(val)))
		}
	}

	if len(whereParts) != len(keyInfo.Columns) {
		// Should match if we didn't return early due to missing key values
		logrus.Warnf("生成 DELETE SQL 失败: 表 %s.%s WHERE 子句部分数量 %d 与键列数 %d 不匹配。",
			schemaName, tableName, len(whereParts), len(keyInfo.Columns))
		return ""
	}

	whereClause := strings.Join(whereParts, " AND ")
	return fmt.Sprintf("DELETE FROM `%s`.`%s` WHERE %s;",
		schemaName, tableName, whereClause)
}

// generateUpdateSQL 生成更新SQL (修改以处理键值和更新值格式化)
func generateUpdateSQL(schemaName, tableName string, keyInfo *db.KeyInfo, diff models.RowDiff) string {
	setClause := ""
	if len(diff.DiffColumns) == 0 {
		// No differing columns, no UPDATE needed
		return ""
	}

	// Build SET clause using differing columns from diff.DiffColumns
	// Use SourceValue from the DiffColumns as the value to set on the target to fix the difference.
	for col, colDiff := range diff.DiffColumns {
		if setClause != "" {
			setClause += ", "
		}

		// Use escapeSQLValue to format the source value for the SET clause
		sourceValue := colDiff.SourceValue
		setClause += fmt.Sprintf("`%s` = %s", col, escapeSQLValue(sourceValue))
	}

	// Build WHERE clause using key values from diff.Key
	var whereParts []string
	if keyInfo == nil || len(keyInfo.Columns) == 0 {
		logrus.Error("generateUpdateSQL: 键信息无效")
		return ""
	}

	if len(diff.Key) == 0 {
		// Key map is empty, cannot build WHERE clause
		logrus.Warnf("生成 UPDATE SQL 失败: 表 %s.%s 差异行没有键值来构建 WHERE 子句。差异详情: %+v",
			schemaName, tableName, diff)
		return ""
	}

	// Build WHERE clause using key columns from keyInfo and values from diff.Key map
	for _, col := range keyInfo.Columns {
		val, exists := diff.Key[col]
		if !exists {
			// Warning: Value for a key column is missing in the diff.Key map!
			logrus.Warnf("生成 UPDATE SQL 失败: 表 %s.%s 差异键数据中缺少键列 '%s' 的值。键数据: %+v",
				schemaName, tableName, col, diff.Key)
			return "" // Cannot generate valid SQL
		}
		// Format key value using escapeSQLValue for SQL literal
		if val == nil {
			whereParts = append(whereParts, fmt.Sprintf("`%s` IS NULL", col))
			// No argument needed for IS NULL
		} else {
			whereParts = append(whereParts, fmt.Sprintf("`%s` = %s", col, escapeSQLValue(val)))
		}
	}

	if len(whereParts) != len(keyInfo.Columns) {
		// Should match if we didn't return early due to missing key values
		logrus.Warnf("生成 UPDATE SQL 失败: 表 %s.%s WHERE 子句部分数量 %d 与键列数 %d 不匹配。",
			schemaName, tableName, len(whereParts), len(keyInfo.Columns))
		return ""
	}

	whereClause := strings.Join(whereParts, " AND ")
	return fmt.Sprintf("UPDATE `%s`.`%s` SET %s WHERE %s;",
		schemaName, tableName, setClause, whereClause)
}

// escapeSQLValue 转义SQL值，用于安全地插入或更新到 SQL 语句中。
// 优化点：更全面地处理各种 Go 类型到 SQL 字面量的转换和转义，特别是二进制和字符串。
// Note: This should handle the types returned by db.GetScannedValue
func escapeSQLValue(val interface{}) string {
	if val == nil {
		return "NULL" // Go's nil maps to SQL's NULL
	}

	// Based on the types returned by db.GetScannedValue:
	switch v := val.(type) {
	case string:
		// String: wrap in single quotes, escape internal single quotes.
		escapedStr := strings.ReplaceAll(v, "'", "''")
		return fmt.Sprintf("'%s'", escapedStr)
	case int64: // Handles all Go signed int types scanned by sql.NullInt64
		return fmt.Sprintf("%d", v)
	case float64: // Handles all Go float types scanned by sql.NullFloat64
		// Use a format that retains precision, e.g., %g or %.10f
		// For float comparison in SQL, exact match can be tricky. Repair should use source value.
		// Using %.10f provides a reasonable representation.
		return fmt.Sprintf("%.10f", v) // Use sufficient precision
	case bool: // Handles bool scanned by sql.NullBool
		if v {
			return "1"
		}
		return "0"
	case time.Time: // Handles time.Time scanned by sql.NullTime
		// Format as SQL datetime string, including milliseconds if possible.
		// Use a standard layout compatible with MySQL.
		// Note: Timezone might need careful handling if DBs have different timezones or use session timezone.
		// Formatting to UTC first might be safer for consistency if storing UTC, but
		// here we format the time object as scanned (which might be local or UTC depending on driver/DSN parseTime setting).
		// Use a format that supports up to microseconds (like MySQL DATETIME/TIMESTAMP supports).
		// "2006-01-02 15:04:05.999999" is a good layout for Go.
		// Ensure the time is not zero value, which can cause issues in SQL.
		if v.IsZero() {
			return "NULL" // Represent zero time as NULL, assuming this is appropriate
		}
		return fmt.Sprintf("'%s'", v.Format("2006-01-02 15:04:05.999999"))
	case []byte: // Handles BINARY, VARBINARY, BLOB types
		// Binary data: Use UNHEX() function with hex string.
		// UNHEX('') for empty binary.
		if len(v) == 0 {
			return "UNHEX('')"
		}
		// Encode as uppercase hex string
		return fmt.Sprintf("UNHEX('%X')", v)
	// Add other types returned by GetScannedValue if necessary.
	case uint64: // If unsigned integers are scanned directly as uint64 (less common with sql.NullInt64)
		// Note: If unsigned are scanned as string (safer for large values), they fall into the string case.
		return fmt.Sprintf("%d", v)

	default:
		// For other types, log a warning and fallback to default string representation with quotes.
		logrus.Warnf("escapeSQLValue encountered unhandled Go type %T (value: %v), using fmt.Sprintf and quoting.", v, v)
		strValue := fmt.Sprintf("%v", v)
		escapedStr := strings.ReplaceAll(strValue, "'", "''")
		return fmt.Sprintf("'%s'", escapedStr)
	}
}

// isJSON checks if a byte slice is likely JSON data (simplified check)
// This helper doesn't seem used in the final comparison logic, but keeping if needed elsewhere.
func isJSON(data []byte) bool {
	// Simplified check: starts with { or [ and ends with } or ] after trimming spaces.
	if data == nil {
		return false
	}
	str := string(data)
	trimmed := strings.TrimSpace(str)
	return (strings.HasPrefix(trimmed, "{") && strings.HasSuffix(trimmed, "}")) ||
		(strings.HasPrefix(trimmed, "[") && strings.HasSuffix(trimmed, "]"))
}

// convertToMapSlice Helper function - Converts slice of TableRowData to slice of map[string]interface{}
// Used for populating SourceOnlyRows and TargetOnlyRows lists in DiffResult (for compatibility).
// These lists contain the full data of the rows only present on one side.
func convertToMapSlice(rows []*models.TableRowData) []map[string]interface{} {
	logger := logrus.WithField("func", "convertToMapSlice")

	if rows == nil {
		logger.Debug("输入行数据数组为 nil")
		return nil // Return nil if input is nil
	}

	logger.Debugf("开始转换 %d 条行数据", len(rows))
	if len(rows) == 0 {
		logger.Debug("没有行数据需要转换")
		return []map[string]interface{}{} // Return empty slice if input is empty but not nil
	}

	result := make([]map[string]interface{}, 0, len(rows)) // Use 0 initial length, append manually
	for i, row := range rows {
		if row == nil {
			logger.Warnf("第 %d 行数据为 nil，跳过", i+1)
			continue // Skip nil elements in the input slice
		}

		if row.Data == nil {
			logger.Warnf("第 %d 行的数据字段为 nil，跳过", i+1)
			continue // Skip if Data map is nil
		}

		// row.Data should be a map. Since NewTableRowData copies the data,
		// we can use it directly as DiffResult wants map[string]interface{}.
		// Appending row.Data directly means the returned slice contains references
		// to the maps inside the TableRowData objects. This is usually acceptable
		// unless those maps are modified later (which they shouldn't be).
		result = append(result, row.Data) // Append the data map

		// Log sample output in debug mode
		if logrus.GetLevel() >= logrus.DebugLevel && i < 3 { // Log first 3 samples or all in debug
			logger.Debugf("第 %d 行数据转换成功: 列数=%d (示例)", i+1, len(row.Data))
		}
	}

	logger.Debugf("转换完成: 输入 %d 条, 输出 %d 条", len(rows), len(result))
	return result
}

// convertToDiffDetails Helper function - Converts slice of RowDifference to slice of RowDiff
// Used for populating DifferentDetails list in DiffResult (for compatibility).
// These list entries represent rows with matching keys but differing content.
func convertToDiffDetails(rows []*models.RowDifference) []models.RowDiff {
	logger := logrus.WithField("func", "convertToDiffDetails")

	if rows == nil {
		logger.Debug("输入行差异数组为 nil")
		return nil // Return nil if input is nil
	}

	logger.Debugf("开始转换 %d 条行差异数据", len(rows))
	if len(rows) == 0 {
		logger.Debug("没有行差异数据需要转换")
		return []models.RowDiff{} // Return empty slice if input is empty but not nil
	}

	result := make([]models.RowDiff, 0, len(rows)) // Use 0 initial length, append manually
	for i, row := range rows {
		if row == nil {
			logger.Warnf("第 %d 条行差异数据为 nil，跳过", i+1)
			continue // Skip nil elements
		}

		// RowDifference entries in the DifferentRows list are expected to have
		// non-nil SourceRow, non-nil TargetRow, and potentially empty DiffColumns.
		// The check for len(row.DifferentColumns) > 0 is moved below.
		if row.SourceRow == nil || row.TargetRow == nil {
			logger.Errorf("第 %d 条行差异数据不完整 (SourceRow=%t, TargetRow=%t)，跳过转换",
				i+1, row.SourceRow != nil, row.TargetRow != nil)
			continue
		}

		// Extract key map from SourceRow.Data using SourceRow.KeyColumns list
		// Use row.SourceRow.KeyColumns for ordered and correct key column names
		keyMap := make(map[string]interface{}, len(row.SourceRow.KeyColumns))
		for _, col := range row.SourceRow.KeyColumns {
			if val, ok := row.SourceRow.Data[col]; ok {
				keyMap[col] = val
			} else {
				// This should not happen if KeyColumns match Data keys from fetchChunkData
				logger.Errorf("第 %d 条行差异的键列 '%s' 在源行数据中不存在，跳过此差异", i+1, col)
				continue // Cannot form a valid key, skip this RowDiff
			}
		}

		// Check if all key columns were successfully extracted
		if len(keyMap) != len(row.SourceRow.KeyColumns) {
			logger.Errorf("第 %d 条行差异键列提取不完整 (%d/%d)，跳过此差异", i+1, len(keyMap), len(row.SourceRow.KeyColumns))
			continue
		}

		// Copy SourceData and TargetData maps for the RowDiff structure
		// Create copies to prevent potential external modification issues.
		sourceDataCopy := make(map[string]interface{}, len(row.SourceRow.Data))
		for k, v := range row.SourceRow.Data {
			sourceDataCopy[k] = v
		}
		targetDataCopy := make(map[string]interface{}, len(row.TargetRow.Data))
		for k, v := range row.TargetRow.Data {
			targetDataCopy[k] = v
		}

		// Copy DiffColumns map
		diffColumnsCopy := make(map[string]models.ColumnDiff, len(row.DifferentColumns))
		for k, v := range row.DifferentColumns {
			diffColumnsCopy[k] = v // ColumnDiff struct is copyable
		}

		// A RowDifference entry in the DifferentRows list signifies that keys matched but content differed.
		// However, FindDifferences might return an empty map if the difference wasn't in column values
		// (e.g., only CRC calculated on row differed, but column values were byte-equal).
		// We still want to represent this row as 'different' in the list, even if no repair SQL is generated for it.
		// So, we append the RowDiff even if len(diffColumnsCopy) is 0.
		// The check for generating SQL happens later in generateFixSQLs.

		rowDiff := models.RowDiff{
			Key:         keyMap,          // Use the extracted key map
			SourceData:  sourceDataCopy,  // Use the copy
			TargetData:  targetDataCopy,  // Use the copy
			DiffColumns: diffColumnsCopy, // Use the copy (might be empty)
		}

		result = append(result, rowDiff)

		// Log sample output in debug mode
		if logrus.GetLevel() >= logrus.DebugLevel && (i < 3 || len(diffColumnsCopy) == 0) { // Log first 3 or any with 0 diff columns in debug
			logger.Debugf("第 %d 条行差异转换结果: 键='%s', 差异列数=%d",
				i+1, row.Key, len(diffColumnsCopy)) // Log the string key for easy identification
			// Optionally log sample diff columns if any
			// for colName, colDiff := range diffColumnsCopy {
			//     logrus.Debugf("  - Diff Col: %s, Source: %v, Target: %v", colName, colDiff.SourceValue, colDiff.TargetValue)
			// }
		} else if logrus.GetLevel() >= logrus.DebugLevel && i < 3 { // Log first 3 with diff columns
			logger.Debugf("第 %d 条行差异转换结果: 键='%s', 差异列数=%d",
				i+1, row.Key, len(diffColumnsCopy))
		}
	}

	logger.Debugf("转换完成: 输入 %d 条, 输出 %d 条", len(rows), len(result))
	return result
}
